package cn.spark.study.sql;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.hive.HiveContext;

/**
 * row_number()开窗函数
 *
 * @author jun.zhang6
 * @date 2020/11/15
 */
public class RowNumberWindowFunction {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf().setAppName("RowNumberWindowFunction");

        JavaSparkContext sc = new JavaSparkContext(conf);

        HiveContext hiveContext = new HiveContext(sc.sc());

        //创建销售额表,sales表
        hiveContext.sql("DROP TABLE IF EXISTS sales");
        hiveContext.sql("CREATE TABLE IF NOT EXISTS sales ("
                + "product STRING,"
                + "category STRING,"
                + "revenue BIGINT)");
        hiveContext.sql("LOAD DATA "
                + "LOCAL INPATH '/usr/local/spark-study/resources/sales.txt' "
                + "INTO TABLE sales");

        //row_number()开窗函数的作用，其实就是给每个分组的数据，按照其顺序，打上一个分组内的行号
        DataFrame top3SalesDF = hiveContext.sql(""
                + "SELECT product,category,revenue "
                + "FROM ("
                + "SELECT "
                + "product,"
                + "category,"
                + "revenue,"
                // row_number()开窗函数的语法说明
                // 首先可以，在SELECT查询时，使用row_number()函数
                // 其次，row_number()函数后面先跟上OVER关键字
                // 然后括号中，是PARTITION BY，也就是说根据哪个字段进行分组
                // 其次是可以用ORDER BY进行组内排序
                // 然后row_number()就可以给每个组内的行，一个组内行号
                + "row_number() OVER (PARTITION BY category ORDER BY revenue DESC) rank "
                + "FROM sales "
                + ") tmp_sales "
                + "WHERE rank<=3");

        //将排名前3的数据，保存到一张表中
        hiveContext.sql("DROP TABLE IF EXISTS top3_sales");

        top3SalesDF.saveAsTable("top3_sales");

        sc.close();
    }
}
